package com.atguigu.edu.realtime.app.dws.user;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.atguigu.edu.realtime.app.BaseAppV1;
import com.atguigu.edu.realtime.bean.UserBackBean;
import com.atguigu.edu.realtime.common.Constant;
import com.atguigu.edu.realtime.common.KafkaTopicConfig;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwdUserUserBackWindow extends BaseAppV1 {
    public static void main(String[] args) {
        new DwdUserUserBackWindow().init(3075,
                2,
                "DwdUserUserBackWindow",
                KafkaTopicConfig.DWD_TRAFFIC_UNIQUE_VISITOR_DETAIL_TOPIC
                );
    }


    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        //使用状态记录用户上次登录时间，找出回流用户
        SingleOutputStreamOperator<UserBackBean> beanStream = findBackUser(stream);
        //开窗聚合
        SingleOutputStreamOperator<UserBackBean> result = winAgg(beanStream);

        //写出到Doris
        writeToDoris(result);
    }

    private void writeToDoris(SingleOutputStreamOperator<UserBackBean> result) {
        result.map(bean -> {
            SerializeConfig conf = new SerializeConfig();
            conf.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;
            System.out.println(bean);
            return JSON.toJSONString(bean, conf);
        }).addSink(FlinkSinkUtil.getDorisSink("edu.dws_user_user_back_window"));

    }

    private SingleOutputStreamOperator<UserBackBean> winAgg(SingleOutputStreamOperator<UserBackBean> beanStream) {
        return beanStream
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<UserBackBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((bean,ts) -> bean.getTs())
                )
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<UserBackBean>() {
                    @Override
                    public UserBackBean reduce(UserBackBean b1, UserBackBean b2) throws Exception {
                        b1.setBackCt(b1.getBackCt() + b2.getBackCt());
                        b1.setUuCt(b1.getUuCt() + b2.getUuCt());
                        return b1;
                    }
                }, new ProcessAllWindowFunction<UserBackBean, UserBackBean, TimeWindow>() {
                    @Override
                    public void process(Context ctx, Iterable<UserBackBean> iterable, Collector<UserBackBean> out) throws Exception {
                        UserBackBean bean = iterable.iterator().next();
                        bean.setStt(DateFormatUtil.toDateTimeString(ctx.window().getStart()));
                        bean.setEdt(DateFormatUtil.toDateTimeString(ctx.window().getEnd()));
                        bean.setCurDate(DateFormatUtil.toDateString(ctx.window().getStart()));

                        out.collect(bean);
                    }
                });
    }

    private SingleOutputStreamOperator<UserBackBean> findBackUser(DataStreamSource<String> stream) {
        return stream
                .map(JSON::parseObject)
                .keyBy(obj -> obj.getJSONObject("common").getString("uid"))
                .process(new KeyedProcessFunction<String, JSONObject, UserBackBean>() {

                    private ValueState<String> lastLoginState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        lastLoginState = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastLoginState", String.class));
                    }

                    @Override
                    public void processElement(JSONObject obj, Context ctx, Collector<UserBackBean> out) throws Exception {
                        String lastDate = lastLoginState.value();
                        Long ts = obj.getLong("ts");
                        String currDate = DateFormatUtil.toDateString(ts);

                        Long uuCt = 0L;
                        Long backCt = 0L;

                        if(!currDate.equals(lastDate)){
                            lastLoginState.update(currDate);
                            uuCt = 1L;
                            if (lastDate != null) {
                                Long last = DateFormatUtil.dateStringToTs(lastDate);
                                Long curr = DateFormatUtil.dateStringToTs(currDate);
                                if((curr -last) / 1000 / 60 / 60 / 24 > Constant.USER_BACK_FLAG_DAY){
                                    backCt = 1L;
                                }
                            }
                        }

                        if(uuCt == 1){
                            out.collect(new UserBackBean("","","",backCt,uuCt,ts));
                        }

                    }
                });

    }
}


